#ifndef RT_USE_OLD_CONN_WRAP

#include "RtConnRlbTcpSendBuffer.h"
#include "RtConnRlbTcp.h"

#define RLBTCP_ENABLE_PDU_POOL

#define RT_CONN_BUFFER_MAX_SIZE_DEFAULT 8192 /// in bytes. 

RlbTcpPduDataPoolType CRtConnRlbTcpSendBuffer::s_pduDataPool;

CRtConnRlbTcpSendBuffer::CRtConnRlbTcpSendBuffer(CRtConnRlbTcp* inConn)
: m_conn(inConn)
, m_maxSize(RT_CONN_BUFFER_MAX_SIZE_DEFAULT)
, m_curSize(0)
, m_seq(0)
, m_encodedData(NULL)
{

}

CRtConnRlbTcpSendBuffer::~CRtConnRlbTcpSendBuffer()
{
	Release();
}

void CRtConnRlbTcpSendBuffer::AddData(
		CRtMessageBlock* inData,
		DWORD inDataLength,
		IRtTransport::DataPriority inPriority, 
		BOOL inNeedAck)
{
	BYTE pduType = RT_BIT_DISABLED(m_conn->m_connType, CRtConnectionManager::CTYPE_PDU_ACK)
		? RT_CONN_RLBTCP_PDU_DATA : ( inNeedAck ? RT_CONN_RLBTCP_PDU_DATA_NEED_ACK : RT_CONN_RLBTCP_PDU_DATA_WITH_ACK );

#ifdef RLBTCP_ENABLE_PDU_POOL
	CRtConnRlbTcpPduData* dataPdu = s_pduDataPool.New();
	dataPdu->Initialize(pduType, inData, inDataLength);
	m_sendDataSet[inPriority].Push(dataPdu);
#else
	m_sendDataSet[inPriority].Push( new CRtConnRlbTcpPduData( pduType, inData, inDataLength ) );
#endif
	m_curSize += inDataLength;
}

RtResult CRtConnRlbTcpSendBuffer::SendBuffer()
{
	int i = IRtTransport::DP_MEDIUM;
	CRtConnRlbTcpPduData* dataPdu = NULL;
#ifdef RLBTCP_ENABLE_DATA_PRIORITY
	for (i=IRtTransport::DP_SYSTEM; i<IRtTransport::DP_NUMBER_OF_PRIORITIES; ++i)
	{
#endif
		while ( !m_sendDataSet[i].IsEmpty() )
		{
			dataPdu = m_sendDataSet[i].Pop();
			BYTE pduType = dataPdu->GetType();
// 			if (pduType!=RT_CONN_RLBTCP_PDU_DATA)
// 			{
				dataPdu->SetAck(m_conn->m_ackToPeer);
				if (pduType==RT_CONN_RLBTCP_PDU_DATA_NEED_ACK)
				{
					++m_seq;
					m_resendQueue.Push(dataPdu);
				}
// 			}
			CRtMessageBlock mb(CRtConnRlbTcpPduBase::GetFixLength(pduType) 
#ifdef RLBTCP_ENABLE_SHORT_DATALEN
				+ (dataPdu->GetShortLength()==0xFF ? sizeof(DWORD) : 0)
#endif
				);
			dataPdu->Encode(mb);
			if (m_encodedData==NULL)
			{
				m_conn->m_unackedDataLen = 0;
				m_encodedData = mb.DuplicateChained();
			}
			else
				m_encodedData->Append(mb.DuplicateChained());

			if (pduType!=RT_CONN_RLBTCP_PDU_DATA_NEED_ACK)
			{
				m_curSize -= dataPdu->GetLength();
#ifdef RLBTCP_ENABLE_PDU_POOL
				s_pduDataPool.Delete(dataPdu);
#else
				delete dataPdu;
#endif
			}
		}
#ifdef RLBTCP_ENABLE_DATA_PRIORITY
	}
#endif

	return SendEncodedData();
}

RtResult CRtConnRlbTcpSendBuffer::SendEncodedData()
{
	if (m_encodedData == NULL)
		return RT_OK;
	
	RtResult rv = m_conn->m_lowerTransport->SendData(*m_encodedData);
	if (RT_FAILED(rv))
	{
		m_encodedData = m_encodedData->ReclaimGarbage();
		return rv;
	}
	m_encodedData->DestroyChained();
	m_encodedData = NULL;
	return RT_OK;
}

void CRtConnRlbTcpSendBuffer::SendKeepAlive()
{
	m_conn->m_unackedDataLen = 0;
	CRtConnRlbTcpPduKeepalive keepalivePdu(m_conn->m_ackToPeer);
	CRtMessageBlock mb(CRtConnRlbTcpPduBase::GetFixLength(RT_CONN_RLBTCP_PDU_KEEPALIVE));
	keepalivePdu.Encode(mb);
	if (m_encodedData==NULL)
	{
		m_encodedData = mb.DuplicateChained();
		SendEncodedData();
		return;
	}

	m_encodedData->Append(mb.DuplicateChained());
}

void CRtConnRlbTcpSendBuffer::SendRtt(DWORD inTimestamp)
{
	CRtConnRlbTcpPduRtt rttPdu(inTimestamp);
	CRtMessageBlock mb(CRtConnRlbTcpPduBase::GetFixLength(RT_CONN_RLBTCP_PDU_RTT));
	rttPdu.Encode(mb);
	if (m_encodedData==NULL)
	{
		m_encodedData = mb.DuplicateChained();
		SendEncodedData();
		return;
	}

	m_encodedData->Append(mb.DuplicateChained());
}

BOOL CRtConnRlbTcpSendBuffer::OnRecvAck(SeqType inAck)
{
	DWORD sizeBefore = m_resendQueue.Size();
	if (sizeBefore==0)
		goto end;

	{
		RT_ASSERTE(sizeBefore<=RT_CONN_SEQ_MAX_VALUE); /// ResendDataQueue size should not be larger than max value of SeqType.
		SeqType rmNum = (SeqType)sizeBefore - (m_seq - inAck);

		for (SeqType i=0; i<rmNum; ++i)
		{
			CRtConnRlbTcpPduData* dataPdu = m_resendQueue.Pop();
			m_curSize -= dataPdu->GetLength();
#ifdef RLBTCP_ENABLE_PDU_POOL
			s_pduDataPool.Delete(dataPdu);
#else
			delete dataPdu;
#endif
		}
	}
	
end:
	return HaveSpace();
}

void CRtConnRlbTcpSendBuffer::OnReconnect(SeqType inAck)
{
	RT_ASSERTE(m_encodedData==NULL);
	OnRecvAck(inAck);
	m_seq = inAck;

	/// Send ResendQueue data.
	CRtConnRlbTcpPduData* dataPdu = NULL;
	DWORD totalSize = m_resendQueue.Size();
	for (DWORD i=0; i<totalSize; ++i)
	{
		dataPdu = m_resendQueue[i];
		dataPdu->SetAck(m_conn->m_ackToPeer);
		++m_seq;

		CRtMessageBlock mb(CRtConnRlbTcpPduBase::GetFixLength(dataPdu->GetType()) 
#ifdef RLBTCP_ENABLE_SHORT_DATALEN
			+ (dataPdu->GetShortLength()==0xFF ? sizeof(DWORD) : 0)
#endif
			);
		dataPdu->Encode(mb);
		if (m_encodedData==NULL)
			m_encodedData = mb.DuplicateChained();
		else
			m_encodedData->Append(mb.DuplicateChained());
	}

	RtResult rv = SendEncodedData();
	if (RT_FAILED(rv))
		return;

	/// Send data set.
	SendBuffer();
}

void CRtConnRlbTcpSendBuffer::Release()
{
	ReleaseEncodedData();

	if (m_curSize==0)
		return; 

	DWORD i, j, totalSize;
	for (i=IRtTransport::DP_SYSTEM; i<IRtTransport::DP_NUMBER_OF_PRIORITIES; ++i)
	{
		totalSize = m_sendDataSet[i].Size();
		for (j=0; j<totalSize; ++j)
		{
#ifdef RLBTCP_ENABLE_PDU_POOL
			s_pduDataPool.Delete(m_sendDataSet[i][j]);
#else
			delete m_sendDataSet[i][j];
#endif
		}
		m_sendDataSet[i].Clear();
	}

	totalSize = m_resendQueue.Size();
	for (i=0; i<totalSize; ++i)
	{
#ifdef RLBTCP_ENABLE_PDU_POOL
		s_pduDataPool.Delete(m_resendQueue[i]);
#else
		delete m_resendQueue[i];
#endif
	}
	m_resendQueue.Clear();
	m_curSize = 0;
	m_seq = 0;
}

void CRtConnRlbTcpSendBuffer::ReleaseEncodedData()
{
	if (m_encodedData)
	{
		m_encodedData->DestroyChained();
		m_encodedData = NULL;
	}
}

#endif // !RT_USE_OLD_CONN_WRAP
